package com.citusdata.elven.feeder;

import com.citusdata.elven.feeder.CarDataFeeder;
import com.citusdata.elven.feeder.LinearRoadClient;
import com.citusdata.elven.httpRpc.HttpRpcChannel;
import com.citusdata.elven.httpRpc.HttpRpcChannelManager;
import com.citusdata.elven.httpRpc.HttpRpcServer;
import com.citusdata.elven.httpRpc.HttpRpcUrl;
import com.citusdata.elven.linearRoad.LinearRoadService;
import com.citusdata.elven.util.ConfigManager;
import com.google.protobuf.Message;
import org.apache.commons.configuration.Configuration;
import org.apache.log4j.Logger;
import static com.citusdata.elven.feeder.ConcurrentDataFeeder.*;

import java.io.FileReader;
import java.io.FileWriter;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.concurrent.CompletionService;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


// Main class responsible for initializing Http servers and clients, and feeding
// data into the system.
public class HttpRpcDataFeeder {
  private static final Logger logger = Logger.getLogger(HttpRpcDataFeeder.class);
  private static final Configuration config = ConfigManager.getConfig();

  private enum ExecutionMode {
    STAND_ALONE_SERVER, STAND_ALONE_CLIENT, SERVER_CLIENT
  }

  public static ExecutionMode readExecModeConfig() {
    boolean server = config.getBoolean("HttpRpcDataFeeder.StandAloneServer", false);
    boolean client = config.getBoolean("HttpRpcDataFeeder.StandAloneClient", false);

    ExecutionMode mode = ExecutionMode.SERVER_CLIENT;  // default

    if (server && client) {
      logger.warn("Incorrect execution mode config, using defaults");
      mode = ExecutionMode.SERVER_CLIENT;
    } else if (server) {
      mode = ExecutionMode.STAND_ALONE_SERVER;
    } else if (client) {
      mode = ExecutionMode.STAND_ALONE_CLIENT;
    }

    return mode;
  }


  public static HttpRpcServer getHttpRpcServer() {
    int maxQueueSize = config.getInt("HttpRpcServer.MaxQueueSize", 100);
    int threadCount  = config.getInt("HttpRpcServer.ThreadCount", 10);

    ThreadPoolExecutor threadPool
      = new ThreadPoolExecutor(threadCount, threadCount,
                               Long.MAX_VALUE, TimeUnit.NANOSECONDS,
                               new LinkedBlockingQueue(maxQueueSize),
                               Executors.defaultThreadFactory(),
                               new ThreadPoolExecutor.DiscardPolicy());
    
    int port = config.getInt("HttpRpcServer.PortNumber", 8080);
    boolean persistent = config.getBoolean("HttpRpcServer.Persistent", true);

    HttpRpcServer httpServer = new HttpRpcServer(threadPool, port, persistent);

    return httpServer;
  }


  public static ThreadPoolExecutor getThreadPoolExecutor() {
    int threadCount = config.getInt("HttpRpcClient.ThreadCount", 10);
    ThreadPoolExecutor threadPool
      = new ThreadPoolExecutor(threadCount, threadCount,
                               Long.MAX_VALUE, TimeUnit.NANOSECONDS,
                               new LinkedBlockingQueue());

    return threadPool;
  }


  protected static void checkHttpRpcConfig() {
    int serverThread  = config.getInt("HttpRpcServer.ThreadCount", 10);
    int clientChannel = config.getInt("HttpRpcChannelManager.MaxTotalChannels", 10);

    if (serverThread < clientChannel) {
      logger.error("Number of server threads: " + serverThread +
                   " is less than number of open channels: " + clientChannel);
    }
  }


  public static void main(String[] args) {
    checkHttpRpcConfig();
    ExecutionMode execMode = readExecModeConfig();

    if (execMode == ExecutionMode.SERVER_CLIENT) {
      HttpRpcServer httpRpcServer = getHttpRpcServer();
      httpRpcServer.registerService(new LinearRoadService());
      Thread thread = new Thread(httpRpcServer);
      thread.start();

    } else if (execMode == ExecutionMode.STAND_ALONE_SERVER) {
      HttpRpcServer httpRpcServer = getHttpRpcServer();
      httpRpcServer.registerService(new LinearRoadService());
      httpRpcServer.run();  // run() returns when the server is shut down
      return;               // nothing to do after server is shut down

    }  else if (execMode == ExecutionMode.STAND_ALONE_CLIENT) {
      ;  // No need to start a server if running as stand alone client
     
    }

    // Client side code relies on a completion service to process responses
    // received by client threads. This introduces an extra layer of indirection
    // and an extra queue, but in return provides the ability to:
    // (1) Impose some ordering on messages sent to the server. By draining a
    // queue at the beginning of each minute, we ensure that all messages for
    // the previous minute are processed and that the expressway statistics for
    // the previous minute are accurate.
    // (2) Avoid infinite memory growth for response objects. Once the queue is
    // full, we drain it, and therefore limit the total amount of memory used
    // by response objects.
    ThreadPoolExecutor threadPool = getThreadPoolExecutor();
    CompletionService completions = new ExecutorCompletionService(threadPool);

    // Client should have input file to feed in car data
    if (args.length < 1) {
      logger.fatal("Usage: HttpRpcDataFeeder car_file [output_file]");
      System.exit(-1);
    } else {
      logger.info("HttpRpcDataFeeder starting to feed data");
    }

    // Get host and port name to connect to
    String host = config.getString("HttpRpcChannel.HostName", "localhost");
    int    port = config.getInt("HttpRpcChannel.PortNumber", 8080);

    BufferedReader inputStream = null;
    PrintWriter    outputStream = null;

    try {
      String inputFileName  = args[0];
      String outputFileName = "/var/tmp/notifications.out";
      if (args.length == 2) {
        outputFileName = args[1];
      }
      inputStream  = new BufferedReader(new FileReader(inputFileName));
      outputStream = new PrintWriter(new FileWriter(outputFileName));

      int requestCount   = 0;
      int previousMinute = 1;
      String inputLine;

      while ((inputLine = inputStream.readLine()) != null) {
        Message request = CarDataFeeder.parseRequest(inputLine);

        int currentMinute = CarDataFeeder.getRequestSemiMinute(inputLine);
        if (currentMinute > previousMinute) {
          printQueuedResponses(completions, requestCount, outputStream);

          requestCount = 0;
          previousMinute = currentMinute;
        }

        requestCount++;
        LinearRoadClient client = new LinearRoadClient(host, port, request);
        completions.submit(client);
      }

      // Print the remaining responses in the queue
      printQueuedResponses(completions, requestCount, outputStream);

    } catch (IOException ex) {
      logger.error("Caught exception when feeding data", ex);

    } finally {
      logger.info("Finished sending requests, shutting down");

      HttpRpcChannelManager manager = LinearRoadClient.manager();
      HttpRpcChannel channel = manager.getChannel(host, port);
      channel.dispatchUrl(HttpRpcUrl.SHUTDOWN_URL);  // Shutdown server
      manager.releaseChannel(channel);

      manager.shutdown();     // Shutdown open connections
      threadPool.shutdown();  // Shutdown active client threads

      try { 
        if (inputStream != null) {
          inputStream.close();
        }
        if (outputStream != null) {
          outputStream.close();
        }
      } catch (IOException ignore) { }
    }
  }
}